package jk.fox.transport.netty.server;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.EventExecutorGroup;
import jk.fox.common.RpcDict;
import jk.fox.common.payload.message.Message;
import jk.fox.common.payload.message.MessageDispatcher;
import jk.fox.common.payload.response.Response;
import jk.fox.common.payload.request.RpcRequest;
import org.apache.log4j.Logger;

/**
 * @author cuichao
 * @Description: RpcServerHandler
 * @Date: create in 2019/12/12 15:33
 */
@ChannelHandler.Sharable
public class RpcServerHandler extends SimpleChannelInboundHandler<RpcRequest> {
    private static Logger logger = Logger.getLogger(RpcServerHandler.class);

    private EventExecutorGroup executors = new DefaultEventExecutorGroup(16);
    private MessageDispatcher dispatcher;


    public RpcServerHandler(MessageDispatcher dispatcher ){
        this.dispatcher = dispatcher;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        logger.info(channel.remoteAddress()+ "is connect, the channelId is: "+ channel.id());
    }




    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcRequest msg) throws Exception {
        logger.info("received message form "+ctx.channel().remoteAddress()+", the requestId is "+msg.getRequestId());
        executors.submit(() -> {
            Message message = new Message();
            message.setStatus(RpcDict.MessageStatus.Request);
            message.setRequest(msg);
            Message dispatch = dispatcher.dispatch(message);
            Response response =  dispatch.getResponse();
            ctx.writeAndFlush(response);
        });
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.warn(cause);
        ctx.close();
    }
}
